/*
    实现异步工作器
*/
#pragma once

#include "buffer.hpp"         //缓冲区
#include <thread>             //线程库
#include <mutex>              //互斥锁
#include <condition_variable> //条件变量
#include <functional>         //包装器
#include <atomic>             //原子类型

namespace log
{
    // 异步工作器类
    using Functor = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        ASYNC_SAFE,  // 安全状态，表示缓冲区满了就阻塞，避免了资源耗尽的风险
        ASUNC_UNSAFE // 非安全状态，不考虑资源耗尽的情况，可以无限扩容，常用与测试
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
         AsyncLooper(const Functor &cb, AsyncType looper_type = AsyncType::ASUNC_UNSAFE) 
         : _looper_type(looper_type), _stop(false), _thread(std::thread(&AsyncLooper::threadEntry, this)),
                                                                                          _callBack(cb) {}
        ~AsyncLooper() { stop(); }
        void stop() 
        {
            _stop = true;
            _cond_con.notify_all(); // 唤醒所有的工作线程
            _thread.join(); // 等待工作线程退出
        }
        void push(const char *data, size_t len)
        {
            // 1.无限扩容-非安全（极限压力测试的情况下使用）2.固定大小
            std::unique_lock<std::mutex> lock(_mutex);
            // 条件变量控制，若缓冲区剩余空间大小大于数据长度，就可以添加数据
            // 如果是安全状态就把这个代码加上，非安全状态就把这个代码屏蔽
            if (_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.writeAbleSize() >= len; }); // 被唤醒之后，如果缓冲区有剩余空间，则添加数据；否则继续休眠（lambda表达式结果为假）
            // 能够走下来代表满足了条件，可以向缓冲区添加数据
            _pro_buf.push(data, len);
            // 唤醒一个消费者对缓冲区中的数据进行处理
            _cond_con.notify_one();
        }

    private:
        void threadEntry() // 线程入口函数
        {
            while (1)
            {
                // 要为互斥锁设置一个生命周期，将缓冲区交换完毕后就解锁(不对数据的处理过程加锁保护)
                {
                    // 1.判断生产缓冲区里有没有数据，有则交换，无则阻塞
                    std::unique_lock<std::mutex> lock(_mutex);
                    //退出标志被设置，且生产缓冲区无数据，这时候在退出，否则有可能会造成生产缓冲区有数据，但是没有被完全处理                   
                    if (_stop && _pro_buf.empty()) break;
                    // 若退出前被唤醒，或者有数据被唤醒，则返回真，继续向下运行，否则重新陷入休眠
                    _cond_con.wait(lock, [&]()
                    { return !_pro_buf.empty() || _stop; }); //_stop是真表示程序退出，把剩余的数据进行交换

                        // 等待完毕，消费者与生产者进行地址交换
                        _con_buf.swap(_pro_buf);
                    // 2.唤醒生产者
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 3.被唤醒后，对消费缓冲区进行数据处理
                _callBack(_con_buf);
                // 4.初始化消费缓冲区
                _con_buf.reset();
            }
        }

    private:
        Functor _callBack; // 具体对缓冲区数据进行处理的回调函数，由异步工作器使用者传入。
    private:
        AsyncType _looper_type;            // 默认是安全模式
        std::atomic<bool> _stop;           // 让工作器停止标志变成原子性操作，提高了线程安全
        Buffer _pro_buf;                   // 生产缓冲区
        Buffer _con_buf;                   // 消费缓冲区
        std::mutex _mutex;                 // 互斥锁
        std::condition_variable _cond_pro; // 生产者，等待队列的条件变量
        std::condition_variable _cond_con; // 消费者，等待队列的条件变量
        std::thread _thread;               // 异步工作器对应的工作线程
    };
}

